package com.atguigu.flink.chapter11.function;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/9/25 10:36
 */
public class Flink02_Function_Table {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);
        
        DataStreamSource<String> waterSensorStream =
            env.fromElements("hello world", "hello atguigu", "hello hello hello atguigu", "abc");
        
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        
        Table table = tenv.fromDataStream(waterSensorStream, $("line"));
        tenv.createTemporaryView("t", table);
        
        // 1. 在table api中使用
        // 1.1 内联方式使用
        /*table
            // 内连接
            //.joinLateral(call(SplitString.class, $("line")))
            .leftOuterJoinLateral(call(SplitString.class, $("line")))
            .select($("line"), $("word"), $("len"))
            .execute()
            .print();*/
        
        // 1.2 先注册函数, 然后再使用
        tenv.createTemporaryFunction("split", SplitString.class);
        
        /*table
            // 内连接
            //.joinLateral(call(SplitString.class, $("line")))
            .leftOuterJoinLateral(call("split", $("line")))
            .select($("line"), $("word"), $("len"))
            .execute()
            .print();*/
        
        // 2. 在sql中使用
        /*tenv.sqlQuery("select" +
                          " line," +
                          " word," +
                          " len " +
                          "from t " +
                          "join lateral table(split(line)) on true")
            .execute()
            .print();*/
        /*tenv.sqlQuery("select" +
                          " line," +
                          " w1," +
                          " len1 " +
                          "from t " +
                          "join lateral table(split(line)) as T(w1, len1) on true")
            .execute()
            .print();*/
        // select .. from a join b on a.id=b.id
        // select .. from a,b where a.id=b.id
        /*tenv.sqlQuery("select" +
                          " line," +
                          " w1," +
                          " len1 " +
                          "from t " +
                          ", lateral table(split(line)) as T(w1, len1)")
            .execute()
            .print();*/
        tenv.sqlQuery("select" +
                          " line," +
                          " w1," +
                          " len1 " +
                          "from t " +
                          " left join lateral table(split(line)) as T(w1, len1) on true")
            .execute()
            .print();
        
    }
    
    // 告诉flink每列的名字和类型
    @FunctionHint(output = @DataTypeHint("row<word string, len int>"))
    public static class SplitString extends TableFunction<Row> {
        public void eval(String line) {
            if (line.length() < 5) return;
            
            String[] words = line.split(" ");
            
            for (String word : words) {
                Row row = Row.of(word, word.length());
                collect(row);
            }
        }
    }
    
}
/*
"hello world"                   hello  5
                                world  5
                
 "hello hello hello atguigu"   ...
 
 */

